package com.atguigu.chapter11.function;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/7/26 9:11
 */
public class Flink01_Function_Scala_01 {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        
        Table table = tEnv.fromValues("hello", "atguigu", "abc");
        
        // 1. 在tableApi中使用自定义函数
        // 1.1 内联方式方式使用
        /*table
            .select($("f0"), call(MyUpperCase.class, $("f0")).as("u"))
            .execute()
            .print();*/
        // 1.2 注册之后使用
        /*tEnv.createTemporaryFunction("upper", MyUpperCase.class);
        table
            .select($("f0"), call("upper", $("f0")).as("u"))
            .execute()
            .print();*/
        // 2. 在sql中使用
        // 只能先注册再使用
        tEnv.createTemporaryFunction("upper", MyUpperCase.class);
        tEnv.sqlQuery("select f0, upper(f0) u from " + table).execute().print();
        
        
    }
    
    public static class MyUpperCase extends ScalarFunction {
        public String eval(String s) {
            if (s == null) {
                return null;
            }
            return s.toUpperCase();
        }
    }
}
